package dsregistry

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"math/rand"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/go-redis/redis/v8"
)

var redisClient *redis.Client
var clientSyncOnce sync.Once
var RegistryCenterInstance *RedisRegistry

// Init registryRedisClientAddr ip:port
func Init(registryRedisClientAddr, password string) {
	clientSyncOnce.Do(func() {
		redisClient = redis.NewClient(&redis.Options{
			Addr:     registryRedisClientAddr,
			Password: password,
		})
		env := os.Getenv("APP_ENV")
		if env == "" {
			env = "dev"
		}
		RegistryCenterInstance = NewRedisRegistryWithEnv(env)
	})
}

// ServiceMeta 服务元数据
type ServiceMeta struct {
	ID         string            `json:"id"`
	Name       string            `json:"name"`
	Address    string            `json:"address"`
	Port       int               `json:"port"`
	Protocol   string            `json:"protocol"`
	Weight     int               `json:"weight"`
	Tags       []string          `json:"tags"`
	Metadata   map[string]string `json:"metadata"`
	Status     string            `json:"status"`
	RegisterAt int64             `json:"register_at"`
	LastBeat   int64             `json:"last_beat"`
}

type ServiceInstance struct {
	*ServiceMeta
	TTL time.Duration `json:"-"`
}

const (
	StatusUp                 = "UP"
	StatusDown               = "DOWN"
	StatusOutOfService       = "OUT_OF_SERVICE"
	DefaultHeartbeatInterval = 5 * time.Second
	DefaultExpirationTime    = 30 * time.Second
)

// ============================= 注册中心结构 =============================

type RedisRegistry struct {
	env           string
	client        *redis.Client
	ctx           context.Context
	serviceCache  map[string][]*ServiceInstance
	cacheLock     sync.RWMutex
	watchers      map[string][]chan []*ServiceInstance
	watcherLock   sync.RWMutex
	stopCh        chan struct{}
	loadBalancers map[string]LoadBalancer
}

func NewRedisRegistryWithEnv(env string) *RedisRegistry {
	ctx := context.Background()
	reg := &RedisRegistry{
		env:           env,
		client:        redisClient,
		ctx:           ctx,
		serviceCache:  make(map[string][]*ServiceInstance),
		watchers:      make(map[string][]chan []*ServiceInstance),
		stopCh:        make(chan struct{}),
		loadBalancers: make(map[string]LoadBalancer),
	}

	reg.loadBalancers["random"] = &RandomLoadBalancer{}
	reg.loadBalancers["roundRobin"] = &RoundRobinLoadBalancer{}
	reg.loadBalancers["weighted"] = &WeightedRoundRobinLoadBalancer{}

	go reg.refreshCache()
	go reg.cleanupExpiredServices()

	return reg
}

// ========== 注册/注销 ==========

func (r *RedisRegistry) Register(instance *ServiceInstance) error {
	if instance.TTL == 0 {
		instance.TTL = DefaultHeartbeatInterval
	}
	instance.Status = StatusUp
	instance.RegisterAt = time.Now().Unix()
	instance.LastBeat = time.Now().Unix()

	data, err := json.Marshal(instance)
	if err != nil {
		return err
	}

	pipe := r.client.TxPipeline()
	pipe.HSet(r.ctx, r.serviceKey(instance.Name), instance.ID, data)
	pipe.ZAdd(r.ctx, r.heartbeatKey(instance.Name), &redis.Z{
		Score:  float64(time.Now().Unix()),
		Member: instance.ID,
	})
	pipe.Expire(r.ctx, r.serviceKey(instance.Name), DefaultExpirationTime)
	pipe.Expire(r.ctx, r.heartbeatKey(instance.Name), DefaultExpirationTime)
	_, err = pipe.Exec(r.ctx)
	if err != nil {
		return err
	}
	go r.startHeartbeat(instance)
	return nil
}

func (r *RedisRegistry) Deregister(serviceName, instanceID string) error {
	pipe := r.client.TxPipeline()
	pipe.HDel(r.ctx, r.serviceKey(serviceName), instanceID)
	pipe.ZRem(r.ctx, r.heartbeatKey(serviceName), instanceID)
	_, err := pipe.Exec(r.ctx)
	if err != nil {
		return err
	}
	r.updateLocalCache(serviceName)
	return nil
}

// ========== 服务发现 ==========

func (r *RedisRegistry) Discover(serviceName string) ([]*ServiceInstance, error) {
	r.cacheLock.RLock()
	if instances, ok := r.serviceCache[serviceName]; ok {
		r.cacheLock.RUnlock()
		return filterHealthyInstances(instances), nil
	}
	r.cacheLock.RUnlock()

	instances, err := r.loadInstancesFromRedis(serviceName)
	if err != nil {
		return nil, err
	}

	r.cacheLock.Lock()
	r.serviceCache[serviceName] = instances
	r.cacheLock.Unlock()
	return filterHealthyInstances(instances), nil
}

func (r *RedisRegistry) SelectInstance(serviceName, strategy string) (*ServiceInstance, error) {
	instances, err := r.Discover(serviceName)
	if err != nil || len(instances) == 0 {
		return nil, errors.New("no available instances")
	}
	lb, ok := r.loadBalancers[strategy]
	if !ok {
		lb = r.loadBalancers["random"]
	}
	return lb.Select(instances), nil
}

func filterHealthyInstances(instances []*ServiceInstance) []*ServiceInstance {
	var result []*ServiceInstance
	for _, inst := range instances {
		if inst.Status == StatusUp {
			result = append(result, inst)
		}
	}
	return result
}

// ========== 监听机制 ==========

func (r *RedisRegistry) Watch(serviceName string) <-chan []*ServiceInstance {
	ch := make(chan []*ServiceInstance, 10)
	r.watcherLock.Lock()
	r.watchers[serviceName] = append(r.watchers[serviceName], ch)
	r.watcherLock.Unlock()
	return ch
}

func (r *RedisRegistry) notifyWatchers(serviceName string, instances []*ServiceInstance) {
	r.watcherLock.RLock()
	defer r.watcherLock.RUnlock()
	for _, ch := range r.watchers[serviceName] {
		select {
		case ch <- instances:
		default:
		}
	}
}

// ========== 心跳机制 ==========

func (r *RedisRegistry) startHeartbeat(instance *ServiceInstance) {
	ticker := time.NewTicker(instance.TTL)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			err := r.updateHeartbeat(instance.Name, instance.ID)
			if err != nil {
				r.markInstanceStatus(instance.Name, instance.ID, StatusDown)
			} else {
				r.markInstanceStatus(instance.Name, instance.ID, StatusUp)
			}
		case <-r.stopCh:
			return
		}
	}
}

func (r *RedisRegistry) updateHeartbeat(serviceName, instanceID string) error {
	heartbeatKey := r.heartbeatKey(serviceName)
	_, err := r.client.ZAdd(r.ctx, heartbeatKey, &redis.Z{
		Score:  float64(time.Now().Unix()),
		Member: instanceID,
	}).Result()
	if err != nil {
		return err
	}
	_, err = r.client.Expire(r.ctx, heartbeatKey, DefaultExpirationTime).Result()
	return err
}

func (r *RedisRegistry) markInstanceStatus(serviceName, instanceID, status string) {
	serviceKey := r.serviceKey(serviceName)
	script := `
	local key = KEYS[1]
	local id = ARGV[1]
	local status = ARGV[2]
	local instance = redis.call('HGET', key, id)
	if not instance then return 0 end
	local data = cjson.decode(instance)
	data['status'] = status
	redis.call('HSET', key, id, cjson.encode(data))
	return 1
	`
	_, _ = r.client.Eval(r.ctx, script, []string{serviceKey}, instanceID, status).Result()
	r.updateLocalCache(serviceName)
}

// ========== 清理过期实例 ==========

func (r *RedisRegistry) cleanupExpiredServices() {
	ticker := time.NewTicker(DefaultExpirationTime / 2)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			r.cleanup()
		case <-r.stopCh:
			return
		}
	}
}

func (r *RedisRegistry) cleanup() {
	now := time.Now().Unix()
	expiredBefore := now - int64(DefaultExpirationTime.Seconds())
	keys, _ := r.client.Keys(r.ctx, r.heartbeatKey("*")).Result()
	for _, key := range keys {
		serviceName := strings.TrimPrefix(key, "registry:heartbeat:")
		expired, err := r.client.ZRangeByScore(r.ctx, key, &redis.ZRangeBy{
			Min: "-inf",
			Max: fmt.Sprintf("%d", expiredBefore),
		}).Result()
		if err != nil || len(expired) == 0 {
			continue
		}
		pipe := r.client.TxPipeline()
		serviceKey := r.serviceKey(serviceName)

		// 1. 删除 Hash 中的实例
		pipe.HDel(r.ctx, serviceKey, expired...)

		// 2. 删除 ZSet 中的心跳记录（类型转换）
		members := make([]interface{}, len(expired))
		for i, v := range expired {
			members[i] = v
		}
		pipe.ZRem(r.ctx, key, members...)
		_, err = pipe.Exec(r.ctx)
		if err == nil {
			r.updateLocalCache(serviceName)
		}
	}
}

// ========== 缓存刷新 ==========

func (r *RedisRegistry) refreshCache() {
	ticker := time.NewTicker(15 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			r.refresh()
		case <-r.stopCh:
			return
		}
	}
}

func (r *RedisRegistry) refresh() {
	keys, err := r.client.Keys(r.ctx, r.serviceKey("*")).Result()
	if err != nil {
		return
	}
	for _, key := range keys {
		name := strings.TrimPrefix(key, "registry:service:")
		instances, err := r.loadInstancesFromRedis(name)
		if err != nil {
			continue
		}
		r.cacheLock.Lock()
		r.serviceCache[name] = instances
		r.cacheLock.Unlock()
		r.notifyWatchers(name, instances)
	}
}

// ========== Redis 结构操作 ==========

func (r *RedisRegistry) loadInstancesFromRedis(serviceName string) ([]*ServiceInstance, error) {
	data, err := r.client.HGetAll(r.ctx, r.serviceKey(serviceName)).Result()
	if err != nil {
		return nil, err
	}
	var instances []*ServiceInstance
	for _, val := range data {
		var inst ServiceInstance
		if err := json.Unmarshal([]byte(val), &inst); err == nil {
			instances = append(instances, &inst)
		}
	}
	return instances, nil
}

func (r *RedisRegistry) updateLocalCache(serviceName string) {
	instances, err := r.loadInstancesFromRedis(serviceName)
	if err != nil {
		return
	}
	r.cacheLock.Lock()
	r.serviceCache[serviceName] = instances
	r.cacheLock.Unlock()
	r.notifyWatchers(serviceName, instances)
}

func (r *RedisRegistry) serviceKey(serviceName string) string {
	return fmt.Sprintf("%s:registry:service:%s", r.env, serviceName)
}

func (r *RedisRegistry) heartbeatKey(serviceName string) string {
	return fmt.Sprintf("%s:registry:heartbeat:%s", r.env, serviceName)
}

func (r *RedisRegistry) Close() {
	close(r.stopCh)
	_ = r.client.Close()
}

// ========== 负载均衡策略 ==========

type LoadBalancer interface {
	Select([]*ServiceInstance) *ServiceInstance
}

type RandomLoadBalancer struct{}

func (lb *RandomLoadBalancer) Select(instances []*ServiceInstance) *ServiceInstance {
	if len(instances) == 0 {
		return nil
	}
	return instances[rand.Intn(len(instances))]
}

type RoundRobinLoadBalancer struct {
	index int
	mu    sync.Mutex
}

func (lb *RoundRobinLoadBalancer) Select(instances []*ServiceInstance) *ServiceInstance {
	if len(instances) == 0 {
		return nil
	}
	lb.mu.Lock()
	defer lb.mu.Unlock()
	inst := instances[lb.index%len(instances)]
	lb.index++
	return inst
}

type WeightedRoundRobinLoadBalancer struct {
	currentWeights map[string]int
	mu             sync.Mutex
}

func (lb *WeightedRoundRobinLoadBalancer) Select(instances []*ServiceInstance) *ServiceInstance {
	if len(instances) == 0 {
		return nil
	}

	lb.mu.Lock()
	defer lb.mu.Unlock()

	if lb.currentWeights == nil {
		lb.currentWeights = make(map[string]int)
	}

	var selected *ServiceInstance
	maxX := -1
	total := 0

	for _, inst := range instances {
		if inst.Status != StatusUp {
			continue
		}
		cw := lb.currentWeights[inst.ID] + inst.Weight
		lb.currentWeights[inst.ID] = cw
		total += inst.Weight
		if cw > maxX {
			maxX = cw
			selected = inst
		}
	}
	if selected != nil {
		lb.currentWeights[selected.ID] -= total
	}
	return selected
}
